package com.study.flink.batch;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.JoinOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;

import java.util.ArrayList;

/**
 * @author stephen
 * @create 2019-01-28 21:05
 * @since 1.0.0
 */
public class BatchDemoWithJoin {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        ArrayList<Tuple2<Integer,String>> data1 = new ArrayList<>();
        data1.add(new Tuple2<>(1,"shen"));
        data1.add(new Tuple2<>(2,"zhao"));
        data1.add(new Tuple2<>(3,"xiang"));

        ArrayList<Tuple2<Integer,String>> data2 = new ArrayList<>();
        data2.add(new Tuple2<>(1,"s"));
        data2.add(new Tuple2<>(2,"z"));
        data2.add(new Tuple2<>(3,"x"));

        DataSource<Tuple2<Integer, String>> source1 = env.fromCollection(data1);
        DataSource<Tuple2<Integer, String>> source2 = env.fromCollection(data2);

        JoinOperator.EquiJoin<Tuple2<Integer, String>, Tuple2<Integer, String>, Object> result = source1.join(source2)
                // 第一个数据集中需要进行比较元素的角标
                .where(0)
                // 第二个数据集中需要进行比较元素的角标
                .equalTo(0)
                // map也可以实现
                .with(new JoinFunction<Tuple2<Integer, String>, Tuple2<Integer, String>, Object>() {
                    @Override
                    public Object join(Tuple2<Integer, String> first, Tuple2<Integer, String> second) throws Exception {
                        return new Tuple3<>(first.f0, first.f1, second.f1);
                    }
                });


        result.print();
    }
}
